package person.tavion.springbootkafkademo;

import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerRecordRecoverer;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory,
            KafkaTemplate<Object, Object> template) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        //最大重试三次
        ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
//        设置重试间隔 10秒， 重试次数为 1 次
        BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
        // 失败进入死信，topic和group，都变成${topic}.DLT和${group}.DLT ，如下是进入死信的例子：
        /*
        # bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group myGroup.DLT && bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group myGroup
        GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                 HOST            CLIENT-ID
        myGroup.DLT     myTopic.DLT     0          4               4               0               test-0-faf8afcc-e4b1-4aca-b064-356163564495 /192.168.3.3    test-0

        GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                 HOST            CLIENT-ID
        myGroup         myTopic         0          800020          800020          0               test-0-968dad79-aebe-4c36-827c-fc8479f988ca /192.168.3.3    test-0
        myGroup         myTopic         1          400012          400012          0               test-0-968dad79-aebe-4c36-827c-fc8479f988ca /192.168.3.3    test-0
        myGroup         myTopic         2          300000          300000          0               test-0-968dad79-aebe-4c36-827c-fc8479f988ca /192.168.3.3    test-0
         */
        factory.setCommonErrorHandler(new DefaultErrorHandler(recoverer,backOff));
        return factory;
    }
}
